<?php
/**
 * 消费者
 * */

namespace App\Lib;

use AMQPConnection;
use AMQPChannel;
use AMQPExchange;
use AMQPQueue;

use App\Http\Model\Hint;
use App\Http\Services\SendGiftServices;

class ConsumerNewLuckGift
{
    const QUEUE_ROUTE = 'global.newluckgift.consumer';// 队列路由键

    protected $queueConfig = [];// 队列总配置
    protected $queueRoute = [];// 队列路由
    protected $connectConfig = [];// 连接配置

    protected $consumerMode = 'callback';// 默认阻塞回调

    public function __construct()
    {
        $this->queueConfig = config('queue') ?? [];
        $this->connectConfig = [
            'host' => $this->queueConfig['RABBITMQ_HOST'] ?? '127.0.0.1',
            'port' => $this->queueConfig['RABBITMQ_PORT'] ?? 5672,
            'vhost' => $this->queueConfig['RABBITMQ_VHOST'] ?? '/',
            'login' => $this->queueConfig['RABBITMQ_LOGIN'] ?? 'guest',
	    'password' => $this->queueConfig['RABBITMQ_PASSWORD'] ?? 'guest',
//	                'prefetch_count' => 20,
			            'heartbeat' => 60,
        ];
    }

    public function consumer()
    {
        $code = Hint::SUCCESS;
        $msg = Hint::getErrCodeMsg($code);
	$data = [];
	$trace = '';

        do {
            try {
                // 队列配置项 路由配置 交换类型
                if (empty($this->queueConfig) or empty($this->queueConfig['queueRoute'])) {
                    $code = Hint::ERR_FAIL;
                    $msg = '队列配置项初始' . Hint::getErrCodeMsg($code);
                    break;
                }

                if (!array_key_exists(self::QUEUE_ROUTE, $this->queueConfig['queueRoute'])) {
                    $code = Hint::ERR_PARAMS_CHECK_FAIL;
                    $msg = '队列路由:' . Hint::getErrCodeMsg($code);
                    break;
                }

                // TODO 队列路由
                $this->queueRoute = $this->queueConfig['queueRoute'][self::QUEUE_ROUTE];

                $conn = new AMQPConnection($this->connectConfig);
                $conn->connect();
                if (!$conn->isConnected()) {
                    $code = Hint::ERR_FAIL;
                    $msg = 'rabbit-mq Connection failed !';
                    break;
                }

                $channel = new AMQPChannel($conn);
                if (!$channel->isConnected()) {
                    $code = Hint::ERR_FAIL;
                    $msg = 'rabbit-mq Connection through channel failed !';
                    break;
                }

                $exchange = new AMQPExchange($channel);
                $exchange->setFlags(AMQP_PASSIVE);//声明一个已存在的交换器的，如果不存在将抛出异常，这个一般用在consume端
                $exchange->setName($this->queueRoute['exchangeName'] ?: '');
                $exchange->setType(AMQP_EX_TYPE_TOPIC); // topic 交换类型
                $exchange->declareExchange();

                //$channel->startTransaction();

                $queue = new AMQPQueue($channel);
                $queue->setName($this->queueRoute['queueName'] ?: '');
                $queue->setFlags(AMQP_DURABLE);
                $queue->declareQueue();

                //绑定
                foreach ($this->queueRoute['routeKey'] as $value) {
                    $queue->bind($this->queueRoute['exchangeName'], $value);
                }

                if ($this->consumerMode == 'callback') {
                    $queue->consume('App\Lib\ConsumerNewLuckGift::callback');
                } else {
                    write_log('message', 'callback', []);
                    $this->message($queue);
                }

                $conn->disconnect();

            } catch (\Exception $e) {
		    $code = Hint::ERR_SYS_FAIL;
		    $trace = $e->getTraceAsString();
		    $msg = Hint::getErrCodeMsg($code) . ':' . $e->getCode() . $e->getFile() . $e->getLine() . $e->getMessage();
		    break;
            }

	} while (false);

	echo $code . $msg;

	write_log('ConsumerNewLuckGift', 'ConsumerNewLuckGift', ['code'=>$code, 'msg'=>$msg, 'trace' => $trace]);

        return compact('code', 'msg', 'data');
    }

    /**
     * 回调处理数据
     *{"from_user_id":273,"to_user_ids":["380","289"],"room_id":3019,"gift_id":67,"gift_num":10}
     * 第一种消费方式,阻塞模式接收消息,程序一直会卡在此处
     * */
    public static function callback(\AMQPEnvelope $envelope,AMQPQueue $queue)
    {
	    if ($envelope) {
            $rotingKey = $envelope->getRoutingKey();
	        $body = $envelope->getBody();

            $body = json_decode($body, true);

            //新幸运礼物
            if ($rotingKey == 'global.newluckgift'){
                $service = new SendGiftServices();
                $service->newluckgiftHandle($body);
            }

            $queue->ack($envelope->getDeliveryTag());
        } else {
            echo 'No message' . PHP_EOL;
        }
    }


    /**
     * 第二种消费方式,非阻塞
     * */
    public function message($queue)
    {
        $message = $queue->get();
        if (!empty($message)) {
            $body = $message->getBody();

            // todo log
            write_log('consumer', 'message', json_decode($body, true));

            // 此处处理业务逻辑


            $queue->ack($message->getDeliveryTag());    //应答，代表该消息已经消费
        }
    }

}
